package com.luych.toolbox.data.service.job;

import com.luych.toolbox.common.enums.*;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketEventPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketLetterPublisher;
import com.luych.toolbox.common.event.base.service.publisher.WebSocketNoticePublisher;
import com.luych.toolbox.common.feign.intf.base.service.TaskFeign;
import com.luych.toolbox.common.feign.intf.base.service.TaskLogFeign;
import com.luych.toolbox.common.feign.param.base.service.TaskLogParam;
import com.luych.toolbox.common.feign.param.base.service.TaskParam;
import com.luych.toolbox.common.feign.param.gateway.center.WebSocketEvent;
import com.luych.toolbox.common.feign.view.base.service.TaskLogView;
import com.luych.toolbox.common.feign.view.base.service.TaskView;
import com.luych.toolbox.data.service.constant.Constant;
import com.luych.toolbox.data.service.core.*;
import com.luych.toolbox.data.service.dao.DataCompareHistoryColumnDao;
import com.luych.toolbox.data.service.dao.DataCompareHistoryDao;
import com.luych.toolbox.data.service.dao.DataCompareHistoryTableDao;
import com.luych.toolbox.data.service.dao.DataSourceDao;
import com.luych.toolbox.data.service.entity.*;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.transaction.support.TransactionTemplate;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;

public class DataCompareHistoryExecuteJob extends QuartzJobBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataCompareHistoryExecuteJob.class);

    private final BusProperties busProperties;
    private final ApplicationEventPublisher applicationEventPublisher;

    private final TransactionTemplate transactionTemplate;

    private final DataSourceDao dataSourceDao;
    private final DataCompareHistoryDao dataCompareHistoryDao;
    private final DataCompareHistoryTableDao dataCompareHistoryTableDao;
    private final DataCompareHistoryColumnDao dataCompareHistoryColumnDao;

    private final TaskFeign taskFeign;
    private final TaskLogFeign taskLogFeign;

    public DataCompareHistoryExecuteJob(BusProperties busProperties,
                                        ApplicationEventPublisher applicationEventPublisher,
                                        TransactionTemplate transactionTemplate,
                                        DataCompareHistoryDao dataCompareHistoryDao,
                                        DataSourceDao dataSourceDao,
                                        DataCompareHistoryTableDao dataCompareHistoryTableDao,
                                        DataCompareHistoryColumnDao dataCompareHistoryColumnDao,
                                        TaskFeign taskFeign, TaskLogFeign taskLogFeign) {
        this.busProperties = busProperties;
        this.applicationEventPublisher = applicationEventPublisher;
        this.transactionTemplate = transactionTemplate;
        this.dataCompareHistoryDao = dataCompareHistoryDao;
        this.dataSourceDao = dataSourceDao;
        this.dataCompareHistoryTableDao = dataCompareHistoryTableDao;
        this.dataCompareHistoryColumnDao = dataCompareHistoryColumnDao;
        this.taskFeign = taskFeign;
        this.taskLogFeign = taskLogFeign;
    }

    private boolean checkDataTypeEqualBySettings(DataSourceType dataSourceTypeOne, CoreDataType dataTypeOne,
                                                 DataSourceType dataSourceTypeTwo, CoreDataType dataTypeTwo) {
        for (DataTypeSetting setting : Constant.DATA_TYPE_SETTINGS) {
            if (Objects.equals(setting.getDataSourceTypeFrom(), dataSourceTypeOne)
                    && Objects.equals(setting.getDataTypeFrom(), dataTypeOne)
                    && Objects.equals(setting.getDataSourceTypeTo(), dataSourceTypeTwo)
                    && Objects.equals(setting.getDataTypeTo(), dataTypeTwo)) {
                return true;
            }
            if (Objects.equals(setting.getDataSourceTypeFrom(), dataSourceTypeTwo)
                    && Objects.equals(setting.getDataTypeFrom(), dataTypeTwo)
                    && Objects.equals(setting.getDataSourceTypeTo(), dataSourceTypeOne)
                    && Objects.equals(setting.getDataTypeTo(), dataTypeOne)) {
                return true;
            }
        }
        return false;
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        String id = context.getJobDetail().getJobDataMap().getString("id");

        WebSocketLetterPublisher letterPublisher = new WebSocketLetterPublisher(busProperties, applicationEventPublisher);
        WebSocketNoticePublisher noticePublisher = new WebSocketNoticePublisher(busProperties, applicationEventPublisher);
        WebSocketEventPublisher eventPublisher = new WebSocketEventPublisher(busProperties, applicationEventPublisher);

        DataCompareHistory compareHistory = dataCompareHistoryDao.findById(id).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_COMPARE_HISTORY_NOT_FOUND));
        DataSource dataSourceOne = dataSourceDao.findById(compareHistory.getDataSourceOneId()).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_SOURCE_NOT_FOUND));
        DataSource dataSourceTwo = dataSourceDao.findById(compareHistory.getDataSourceTwoId()).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_SOURCE_NOT_FOUND));

        this.startCompareHistory(compareHistory);

        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_COMPARE_RUNNING, compareHistory.getId()));
        TaskView taskResult = this.createTask(compareHistory, dataSourceOne, dataSourceTwo);

        LOGGER.info("start to compare data source [{}] and data source [{}]...", dataSourceOne.getUrl(), dataSourceTwo.getUrl());
        try {
            List<DataCompareHistoryTable> compareTables = new ArrayList<>();
            List<DataCompareHistoryColumn> compareColumns = new ArrayList<>();
            this.doCompare(compareHistory, compareTables, compareColumns, dataSourceOne, dataSourceTwo, taskResult);
            this.createTaskLog(taskResult.getId(), "start to save compare result ...");
            this.saveToDB(compareTables, compareColumns, compareHistory);
            this.createTaskLog(taskResult.getId(), "success to save compare result ...");
            this.onSuccess(compareHistory, taskResult);
            LOGGER.info("complete to compare data source [{}] and data source [{}]...", dataSourceOne.getUrl(), dataSourceTwo.getUrl());
        } catch (Exception e) {
            LOGGER.info("error to compare data source [{}] and data source [{}]: {}", dataSourceOne.getUrl(), dataSourceTwo.getUrl(), e.getMessage(), e);
            onError(compareHistory, taskResult, e);
        }
    }

    private void startCompareHistory(DataCompareHistory compareHistory) {
        compareHistory.setStartDate(new Date());
        compareHistory.setStatus(DataCompareHistoryStatus.RUNNING);
        dataCompareHistoryDao.save(compareHistory);
    }

    private TaskView createTask(DataCompareHistory compareHistory, DataSource dataSourceOne, DataSource dataSourceTwo) {
        TaskParam taskParam = new TaskParam();
        taskParam.setType(TaskType.DATA_COMPARE);
        taskParam.setStatus(TaskStatus.PENDING);
        taskParam.setName("对比数据源");
        taskParam.setDescription("对比数据源[" + dataSourceOne.getName() + "<->" + dataSourceTwo.getName() + "]");
        taskParam.setOwnerId(compareHistory.getOwnerId());
        TaskView taskResult = taskFeign.create(taskParam);
        taskFeign.start(taskResult.getId());
        return taskResult;
    }

    private TaskLogView createTaskLog(String taskId, String log) {
        LOGGER.info(log);
        TaskLogParam taskLogParam = new TaskLogParam();
        taskLogParam.setTaskId(taskId);
        taskLogParam.setLog(log);
        return taskLogFeign.create(taskLogParam);
    }

    private void doCompare(DataCompareHistory compareHistory,
                           List<DataCompareHistoryTable> compareTables, List<DataCompareHistoryColumn> compareColumns,
                           DataSource dataSourceOne, DataSource dataSourceTwo,
                           TaskView task) {

        this.createTaskLog(task.getId(), "get meta info from " + dataSourceOne.getUrl() + " ...");
        CoreDatabase databaseOne = CoreBuilder.buildConnection(dataSourceOne).getDatabase();
        this.createTaskLog(task.getId(), "get meta info from " + dataSourceTwo.getUrl() + " ...");
        CoreDatabase databaseTwo = CoreBuilder.buildConnection(dataSourceTwo).getDatabase();

        List<String> sortTableNames = this.collectAndSortTableName(databaseOne, databaseTwo);
        this.createTaskLog(task.getId(), "there are " + sortTableNames.size() + " tables in 2 database");

        this.createTaskLog(task.getId(), "start compare tables ...");
        for (String tableName : sortTableNames) {
            Optional<? extends CoreTable> tableOneOpt = databaseOne.getTables().stream().filter(tbl -> tableName.equalsIgnoreCase(tbl.getName())).findFirst();
            Optional<? extends CoreTable> tableTwoOpt = databaseTwo.getTables().stream().filter(tbl -> tableName.equalsIgnoreCase(tbl.getName())).findFirst();
            DataCompareHistoryTable tableCompare = this.createTableCompare(compareHistory, tableOneOpt, tableTwoOpt);
            if (tableOneOpt.isPresent() && tableTwoOpt.isPresent()) {
                CoreTable tableOne = tableOneOpt.get();
                CoreTable tableTwo = tableTwoOpt.get();
                tableCompare.setLevel(DataCompareHistoryTableLevel.SAME);
                if (!Objects.equals(tableCompare.getCommentOne(), tableCompare.getCommentTwo())) {
                    this.setTableCommentDiff(tableCompare, dataSourceOne, tableOne, dataSourceTwo, tableTwo);
                }
                if (!Objects.equals(tableCompare.getPrimaryKeyOne().toUpperCase(), tableCompare.getPrimaryKeyTwo().toUpperCase())) {
                    this.setTablePrimaryKeyDiff(tableCompare, dataSourceOne, tableOne, dataSourceTwo, tableTwo);
                }
            } else if (tableOneOpt.isPresent()) {
                CoreTable tableOne = tableOneOpt.get();
                this.setTableTwoMiss(tableCompare, dataSourceTwo, tableOne);
            } else if (tableTwoOpt.isPresent()) {
                CoreTable tableTwo = tableTwoOpt.get();
                this.setTableOneMiss(tableCompare, dataSourceOne, tableTwo);
            }

            if (tableOneOpt.isPresent() && tableTwoOpt.isPresent()) {
                CoreTable tableOne = tableOneOpt.get();
                CoreTable tableTwo = tableTwoOpt.get();
                List<String> sortColumnNames = this.collectAndSortColumnName(tableOne, tableTwo);
                for (String columnName : sortColumnNames) {
                    Optional<? extends CoreColumn> columnOneOpt = tableOne.getColumns().stream().filter(col -> col.getName().equalsIgnoreCase(columnName)).findFirst();
                    Optional<? extends CoreColumn> columnTwoOpt = tableTwo.getColumns().stream().filter(col -> col.getName().equalsIgnoreCase(columnName)).findFirst();
                    DataCompareHistoryColumn columnCompare = this.createColumnCompare(compareHistory, tableOne, columnOneOpt, tableTwo, columnTwoOpt);
                    if (columnOneOpt.isPresent() && columnTwoOpt.isPresent()) {
                        CoreColumn columnOne = columnOneOpt.get();
                        CoreColumn columnTwo = columnTwoOpt.get();
                        this.setColumnTypeDiff(columnCompare, dataSourceOne, columnOne, dataSourceTwo, columnTwo);
                    } else if (columnOneOpt.isPresent()) {
                        CoreColumn columnOne = columnOneOpt.get();
                        this.setColumnTwoMiss(columnCompare, dataSourceTwo, columnOne);
                    } else if (columnTwoOpt.isPresent()) {
                        CoreColumn columnTwo = columnTwoOpt.get();
                        this.setColumnOneMiss(columnCompare, dataSourceOne, columnTwo);
                    }
                    compareColumns.add(columnCompare);

                    if (!DataCompareHistoryColumnLevel.SAME.equals(columnCompare.getLevel())) {
                        this.setTableColumnDiff(tableCompare, columnCompare);
                    }
                }
            }

            compareTables.add(tableCompare);

            if (!DataCompareHistoryTableLevel.SAME.equals(tableCompare.getLevel())) {
                this.setCompareHistoryDiff(compareHistory, tableCompare);
            }
        }
    }

    private List<String> collectAndSortTableName(CoreDatabase databaseOne, CoreDatabase databaseTwo) {
        Set<String> allTableNames = new HashSet<>();
        allTableNames.addAll(databaseOne.getTables().stream().map(CoreTable::getName).map(String::toUpperCase).collect(Collectors.toList()));
        allTableNames.addAll(databaseTwo.getTables().stream().map(CoreTable::getName).map(String::toUpperCase).collect(Collectors.toList()));
        List<String> sortTableNames = new ArrayList<>(allTableNames);
        sortTableNames.sort(String::compareTo);
        return sortTableNames;
    }

    private List<String> collectAndSortColumnName(CoreTable tableOne, CoreTable tableTwo) {
        Set<String> allColumnNames = new HashSet<>();
        allColumnNames.addAll(tableOne.getColumns().stream().map(CoreColumn::getName).map(String::toUpperCase).collect(Collectors.toList()));
        allColumnNames.addAll(tableTwo.getColumns().stream().map(CoreColumn::getName).map(String::toUpperCase).collect(Collectors.toList()));
        List<String> sortColumnNames = new ArrayList<>(allColumnNames);
        sortColumnNames.sort(String::compareTo);
        return sortColumnNames;
    }

    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
    private DataCompareHistoryTable createTableCompare(DataCompareHistory compareHistory,
                                                       Optional<? extends CoreTable> tableOneOpt,
                                                       Optional<? extends CoreTable> tableTwoOpt) {
        DataCompareHistoryTable tableCompare = new DataCompareHistoryTable();
        tableCompare.setId(UUID.randomUUID().toString());
        tableCompare.setCreateDate(compareHistory.getCreateDate());
        tableCompare.setCreateUserId(compareHistory.getCreateUserId());
        tableCompare.setDataCompareHistoryId(compareHistory.getId());
        tableOneOpt.map(CoreTable::getName).ifPresent(tableCompare::setTableOne);
        tableTwoOpt.map(CoreTable::getName).ifPresent(tableCompare::setTableTwo);
        tableOneOpt.map(CoreTable::getComment).ifPresent(tableCompare::setCommentOne);
        tableTwoOpt.map(CoreTable::getComment).ifPresent(tableCompare::setCommentTwo);
        tableCompare.setPrimaryKeyListOne(tableOneOpt.map(CoreTable::getPrimaryKey).orElse(new ArrayList<>()));
        tableCompare.setPrimaryKeyListTwo(tableTwoOpt.map(CoreTable::getPrimaryKey).orElse(new ArrayList<>()));
        return tableCompare;
    }

    private void setTableOneMiss(DataCompareHistoryTable tableCompare, DataSource dataSourceOne, CoreTable tableTwo) {
        tableCompare.setLevel(DataCompareHistoryTableLevel.MISS);
        tableCompare.setDifference(String.format("数据源%s中缺失此表", dataSourceOne.getName()));

        CoreTable tableOne = CoreBuilder.buildTable(dataSourceOne.getType(), tableTwo);

        tableCompare.addOneToTwo(tableTwo.getRemoveSql(CaseType.UNCHANGE));
        tableCompare.addOneToTwoLower(tableTwo.getRemoveSql(CaseType.LOWER_CASE));
        tableCompare.addOneToTwoUpper(tableTwo.getRemoveSql(CaseType.UPPER_CASE));

        tableCompare.addTwoToOne(tableOne.getCreateSql(CaseType.UNCHANGE));
        tableCompare.addTwoToOneLower(tableOne.getCreateSql(CaseType.LOWER_CASE));
        tableCompare.addTwoToOneUpper(tableOne.getCreateSql(CaseType.UPPER_CASE));
    }

    private void setTableTwoMiss(DataCompareHistoryTable tableCompare, DataSource dataSourceTwo, CoreTable tableOne) {
        tableCompare.setLevel(DataCompareHistoryTableLevel.MISS);
        tableCompare.setDifference(String.format("数据源%s中缺失此表", dataSourceTwo.getName()));

        CoreTable tableTwo = CoreBuilder.buildTable(dataSourceTwo.getType(), tableOne);

        tableCompare.addOneToTwo(tableTwo.getCreateSql(CaseType.UNCHANGE));
        tableCompare.addOneToTwoLower(tableTwo.getCreateSql(CaseType.LOWER_CASE));
        tableCompare.addOneToTwoUpper(tableTwo.getCreateSql(CaseType.UPPER_CASE));

        tableCompare.addTwoToOne(tableOne.getRemoveSql(CaseType.UNCHANGE));
        tableCompare.addTwoToOneLower(tableOne.getRemoveSql(CaseType.LOWER_CASE));
        tableCompare.addTwoToOneUpper(tableOne.getRemoveSql(CaseType.UPPER_CASE));
    }

    private void setTableCommentDiff(DataCompareHistoryTable tableCompare,
                                     DataSource dataSourceOne, CoreTable tableOne,
                                     DataSource dataSourceTwo, CoreTable tableTwo) {
        tableCompare.setLevel(DataCompareHistoryTableLevel.DIFF);
        tableCompare.addDifference("注释不一致");

        CoreTable tableOneInTwo = CoreBuilder.buildTable(dataSourceTwo.getType(), tableOne);
        CoreTable tableTwoInOne = CoreBuilder.buildTable(dataSourceOne.getType(), tableTwo);

        tableCompare.addOneToTwo(tableOneInTwo.getCommentSql(CaseType.UNCHANGE));
        tableCompare.addOneToTwoLower(tableOneInTwo.getCommentSql(CaseType.LOWER_CASE));
        tableCompare.addOneToTwoUpper(tableOneInTwo.getCommentSql(CaseType.UPPER_CASE));

        tableCompare.addTwoToOne(tableTwoInOne.getCommentSql(CaseType.UNCHANGE));
        tableCompare.addTwoToOneLower(tableTwoInOne.getCommentSql(CaseType.LOWER_CASE));
        tableCompare.addTwoToOneUpper(tableTwoInOne.getCommentSql(CaseType.UPPER_CASE));
    }

    private void setTablePrimaryKeyDiff(DataCompareHistoryTable tableCompare,
                                        DataSource dataSourceOne, CoreTable tableOne,
                                        DataSource dataSourceTwo, CoreTable tableTwo) {
        tableCompare.setLevel(DataCompareHistoryTableLevel.DIFF);
        tableCompare.addDifference("主键不一致");

        CoreTable tableOneInTwo = CoreBuilder.buildTable(dataSourceTwo.getType(), tableOne);
        CoreTable tableTwoInOne = CoreBuilder.buildTable(dataSourceOne.getType(), tableTwo);

        if (!tableTwo.getPrimaryKey().isEmpty()) {
            tableCompare.addOneToTwo(tableOneInTwo.getRemovePrimaryKeySql(CaseType.UNCHANGE));
            tableCompare.addOneToTwoLower(tableOneInTwo.getRemovePrimaryKeySql(CaseType.LOWER_CASE));
            tableCompare.addOneToTwoUpper(tableOneInTwo.getRemovePrimaryKeySql(CaseType.UPPER_CASE));
        }
        tableCompare.addOneToTwo(tableOneInTwo.getCreatePrimaryKeySql(CaseType.UNCHANGE));
        tableCompare.addOneToTwoLower(tableOneInTwo.getCreatePrimaryKeySql(CaseType.LOWER_CASE));
        tableCompare.addOneToTwoUpper(tableOneInTwo.getCreatePrimaryKeySql(CaseType.UPPER_CASE));

        if (!tableOne.getPrimaryKey().isEmpty()) {
            tableCompare.addTwoToOne(tableTwoInOne.getRemovePrimaryKeySql(CaseType.UNCHANGE));
            tableCompare.addTwoToOneLower(tableTwoInOne.getRemovePrimaryKeySql(CaseType.LOWER_CASE));
            tableCompare.addTwoToOneUpper(tableTwoInOne.getRemovePrimaryKeySql(CaseType.UPPER_CASE));
        }
        tableCompare.addTwoToOne(tableTwoInOne.getCreatePrimaryKeySql(CaseType.UNCHANGE));
        tableCompare.addTwoToOneLower(tableTwoInOne.getCreatePrimaryKeySql(CaseType.LOWER_CASE));
        tableCompare.addTwoToOneUpper(tableTwoInOne.getCreatePrimaryKeySql(CaseType.UPPER_CASE));
    }

    private void setTableColumnDiff(DataCompareHistoryTable tableCompare, DataCompareHistoryColumn columnCompare) {
        tableCompare.setLevel(DataCompareHistoryTableLevel.DIFF);
        tableCompare.addDifference("字段不一致");

        tableCompare.addOneToTwo(columnCompare.getOneToTwo());
        tableCompare.addOneToTwoLower(columnCompare.getOneToTwoLower());
        tableCompare.addOneToTwoUpper(columnCompare.getOneToTwoUpper());

        tableCompare.addTwoToOne(columnCompare.getTwoToOne());
        tableCompare.addTwoToOneLower(columnCompare.getTwoToOneLower());
        tableCompare.addTwoToOneUpper(columnCompare.getTwoToOneUpper());
    }

    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
    private DataCompareHistoryColumn createColumnCompare(DataCompareHistory compareHistory,
                                                         CoreTable tableOne, Optional<? extends CoreColumn> columnOneOpt,
                                                         CoreTable tableTwo, Optional<? extends CoreColumn> columnTwoOpt) {
        DataCompareHistoryColumn columnCompare = new DataCompareHistoryColumn();
        columnCompare.setId(UUID.randomUUID().toString());
        columnCompare.setCreateDate(compareHistory.getCreateDate());
        columnCompare.setCreateUserId(compareHistory.getCreateUserId());
        columnCompare.setDataCompareHistoryId(compareHistory.getId());
        columnCompare.setTableOne(tableOne.getName());
        columnCompare.setTableTwo(tableTwo.getName());
        columnOneOpt.map(CoreColumn::getName).ifPresent(columnCompare::setColumnOne);
        columnTwoOpt.map(CoreColumn::getName).ifPresent(columnCompare::setColumnTwo);
        columnOneOpt.map(CoreColumn::getColumnType).ifPresent(columnCompare::setTypeOne);
        columnTwoOpt.map(CoreColumn::getColumnType).ifPresent(columnCompare::setTypeTwo);
        columnOneOpt.map(CoreColumn::isNullable).ifPresent(columnCompare::setNullableOne);
        columnTwoOpt.map(CoreColumn::isNullable).ifPresent(columnCompare::setNullableTwo);
        columnOneOpt.map(CoreColumn::isIncrement).ifPresent(columnCompare::setIncrementOne);
        columnTwoOpt.map(CoreColumn::isIncrement).ifPresent(columnCompare::setIncrementTwo);
        columnOneOpt.map(CoreColumn::getComment).ifPresent(columnCompare::setCommentOne);
        columnTwoOpt.map(CoreColumn::getComment).ifPresent(columnCompare::setCommentTwo);
        return columnCompare;
    }

    private void setColumnOneMiss(DataCompareHistoryColumn columnCompare, DataSource dataSourceOne, CoreColumn columnTwo) {
        columnCompare.setLevel(DataCompareHistoryColumnLevel.MISS);
        columnCompare.setDifference(String.format("数据源%s中缺失此字段", dataSourceOne.getName()));

        CoreColumn columnOne = CoreBuilder.buildColumn(dataSourceOne.getType(), columnTwo);

        columnCompare.addOneToTwo(columnTwo.getRemoveSql(CaseType.UNCHANGE));
        columnCompare.addOneToTwoLower(columnTwo.getRemoveSql(CaseType.LOWER_CASE));
        columnCompare.addOneToTwoUpper(columnTwo.getRemoveSql(CaseType.UPPER_CASE));

        columnCompare.addTwoToOne(columnOne.getCreateSql(CaseType.UNCHANGE));
        columnCompare.addTwoToOneLower(columnOne.getCreateSql(CaseType.LOWER_CASE));
        columnCompare.addTwoToOneUpper(columnOne.getCreateSql(CaseType.UPPER_CASE));
    }

    private void setColumnTwoMiss(DataCompareHistoryColumn columnCompare, DataSource dataSourceTwo, CoreColumn columnOne) {
        columnCompare.setLevel(DataCompareHistoryColumnLevel.MISS);
        columnCompare.setDifference(String.format("数据源%s中缺失此字段", dataSourceTwo.getName()));

        CoreColumn columnTwo = CoreBuilder.buildColumn(dataSourceTwo.getType(), columnOne);

        columnCompare.addOneToTwo(columnTwo.getCreateSql(CaseType.UNCHANGE));
        columnCompare.addOneToTwoLower(columnTwo.getCreateSql(CaseType.LOWER_CASE));
        columnCompare.addOneToTwoUpper(columnTwo.getCreateSql(CaseType.UPPER_CASE));

        columnCompare.addTwoToOne(columnOne.getRemoveSql(CaseType.UNCHANGE));
        columnCompare.addTwoToOneLower(columnOne.getRemoveSql(CaseType.LOWER_CASE));
        columnCompare.addTwoToOneUpper(columnOne.getRemoveSql(CaseType.UPPER_CASE));
    }

    private void setColumnTypeDiff(DataCompareHistoryColumn columnCompare,
                                   DataSource dataSourceOne, CoreColumn columnOne,
                                   DataSource dataSourceTwo, CoreColumn columnTwo) {
        columnCompare.setLevel(DataCompareHistoryColumnLevel.SAME);
        if (!Objects.equals(columnOne.getDataType().getName(), columnTwo.getDataType().getName()) &&
                !this.checkDataTypeEqualBySettings(dataSourceOne.getType(), columnOne.getDataType(), dataSourceTwo.getType(), columnTwo.getDataType())) {
            columnCompare.setLevel(DataCompareHistoryColumnLevel.DIFF);
            columnCompare.addDifference("数据类型不一致");
        } else if (columnOne.getDataType().supportCharLength() && columnTwo.getDataType().supportCharLength() &&
                !Objects.equals(columnOne.getCharLength(), columnTwo.getCharLength())) {
            columnCompare.setLevel(DataCompareHistoryColumnLevel.DIFF);
            columnCompare.addDifference("字符长度不一致");
        } else if (columnOne.getDataType().supportNumericPrecision() && columnTwo.getDataType().supportNumericPrecision() &&
                !Objects.equals(columnOne.getNumericPrecision(), columnTwo.getNumericPrecision())) {
            columnCompare.setLevel(DataCompareHistoryColumnLevel.DIFF);
            columnCompare.addDifference("数字精度不一致");
        } else if (columnOne.getDataType().supportNumericScale() && columnTwo.getDataType().supportNumericScale() &&
                !Objects.equals(columnOne.getNumericScale(), columnTwo.getNumericScale())) {
            columnCompare.setLevel(DataCompareHistoryColumnLevel.DIFF);
            columnCompare.addDifference("数字小数位数不一致");
        }
        if (!Objects.equals(columnCompare.getNullableOne(), columnCompare.getNullableTwo())) {
            columnCompare.setLevel(DataCompareHistoryColumnLevel.DIFF);
            columnCompare.addDifference("Nullable不一致");
        }
        if (!Objects.equals(columnCompare.getIncrementOne(), columnCompare.getIncrementTwo())) {
            columnCompare.setLevel(DataCompareHistoryColumnLevel.DIFF);
            columnCompare.addDifference("自增不一致");
        }
        if (!Objects.equals(columnCompare.getCommentOne(), columnCompare.getCommentTwo())) {
            columnCompare.setLevel(DataCompareHistoryColumnLevel.DIFF);
            columnCompare.addDifference("注释不一致");
        }

        if (!DataCompareHistoryColumnLevel.SAME.equals(columnCompare.getLevel())) {
            CoreColumn columnOneInTwo = CoreBuilder.buildColumn(dataSourceTwo.getType(), columnOne);
            CoreColumn columnTwoInOne = CoreBuilder.buildColumn(dataSourceOne.getType(), columnTwo);

            columnCompare.addOneToTwo(columnOneInTwo.getModifySql(CaseType.UNCHANGE));
            columnCompare.addOneToTwoLower(columnOneInTwo.getModifySql(CaseType.LOWER_CASE));
            columnCompare.addOneToTwoUpper(columnOneInTwo.getModifySql(CaseType.UPPER_CASE));

            columnCompare.addTwoToOne(columnTwoInOne.getModifySql(CaseType.UNCHANGE));
            columnCompare.addTwoToOneLower(columnTwoInOne.getModifySql(CaseType.LOWER_CASE));
            columnCompare.addTwoToOneUpper(columnTwoInOne.getModifySql(CaseType.UPPER_CASE));
        }
    }

    private void setCompareHistoryDiff(DataCompareHistory compareHistory, DataCompareHistoryTable tableCompare) {
        compareHistory.addOneToTwo(tableCompare.getOneToTwo());
        compareHistory.addOneToTwoLower(tableCompare.getOneToTwoLower());
        compareHistory.addOneToTwoUpper(tableCompare.getOneToTwoUpper());

        compareHistory.addTwoToOne(tableCompare.getTwoToOne());
        compareHistory.addTwoToOneLower(tableCompare.getTwoToOneLower());
        compareHistory.addTwoToOneUpper(tableCompare.getTwoToOneUpper());
    }

    private void saveToDB(List<DataCompareHistoryTable> compareTables, List<DataCompareHistoryColumn> compareColumns, DataCompareHistory compareHistory) {
        transactionTemplate.executeWithoutResult(status -> {
            dataCompareHistoryTableDao.saveAll(compareTables);
            dataCompareHistoryColumnDao.saveAll(compareColumns);

            compareHistory.setStatus(DataCompareHistoryStatus.SUCCESS);
            compareHistory.setCompleteDate(new Date());
            dataCompareHistoryDao.save(compareHistory);
        });
    }

    private void onSuccess(DataCompareHistory compareHistory, TaskView taskResult) {
        String datetime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date());
        WebSocketLetterPublisher letterPublisher = new WebSocketLetterPublisher(busProperties, applicationEventPublisher);
        WebSocketNoticePublisher noticePublisher = new WebSocketNoticePublisher(busProperties, applicationEventPublisher);
        WebSocketEventPublisher eventPublisher = new WebSocketEventPublisher(busProperties, applicationEventPublisher);

        String content = "数据对比任务[" + compareHistory.getName() + "]于" + datetime + "执行成功！";
        letterPublisher.success(compareHistory.getOwnerId(), "数据对比任务执行通知", content);
        noticePublisher.success(compareHistory.getOwnerId(), content);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_COMPARE_SUCCESS, compareHistory.getId()));

        taskFeign.complete(taskResult.getId(), TaskStatus.SUCCESS, "成功");
    }

    private void onError(DataCompareHistory compareHistory, TaskView taskResult, Exception e) {
        String datetime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date());
        WebSocketLetterPublisher letterPublisher = new WebSocketLetterPublisher(busProperties, applicationEventPublisher);
        WebSocketNoticePublisher noticePublisher = new WebSocketNoticePublisher(busProperties, applicationEventPublisher);
        WebSocketEventPublisher eventPublisher = new WebSocketEventPublisher(busProperties, applicationEventPublisher);

        StringWriter writer = new StringWriter();
        e.printStackTrace(new PrintWriter(writer));

        compareHistory.setStatus(DataCompareHistoryStatus.FAILED);
        compareHistory.setErrorMessage(e.getMessage());
        compareHistory.setErrorTrace(writer.toString());
        dataCompareHistoryDao.save(compareHistory);

        String content = "数据对比任务[" + compareHistory.getName() + "]于" + datetime + "执行异常，请及时处理。";
        noticePublisher.error(compareHistory.getOwnerId(), content);
        content += "\n\n\n错误信息：\n" + e.getMessage() + "\n\n\n异常堆栈：\n" + writer;
        letterPublisher.error(compareHistory.getOwnerId(), "数据对比任务执行通知", content);
        eventPublisher.send(new WebSocketEvent(WebSocketEvent.Type.DATA_COMPARE_ERROR, compareHistory.getId()));

        taskFeign.complete(taskResult.getId(), TaskStatus.ERROR, e.getMessage());
    }
}
